package com.bigdata.quality.config.converter;

import java.util.HashMap;
import java.util.Map;

public class MSSQLColumnTypeConverter implements ColumnTypeConverter {
  public static final MSSQLColumnTypeConverter instance = new MSSQLColumnTypeConverter();

  @Override
  public Map<String, String> toFlinkColumnTypeMap() {
    Map<String, String> columnTypeMap = new HashMap<>();

    // 数值类型映射
    columnTypeMap.put("tinyint", "smallint");
    columnTypeMap.put("smallint", "smallint");
    columnTypeMap.put("int", "int");
    columnTypeMap.put("integer", "int");
    columnTypeMap.put("bigint", "bigint");
    columnTypeMap.put("real", "float");
    columnTypeMap.put("float", "double");
    columnTypeMap.put("decimal", "decimal(38,2)");
    columnTypeMap.put("numeric", "decimal(38,2)");
    columnTypeMap.put("money", "decimal(20,4)");
    columnTypeMap.put("smallmoney", "decimal(20,4)");
    columnTypeMap.put("bit", "boolean");
    columnTypeMap.put("double", "double");

    // 日期时间类型映射
    columnTypeMap.put("date", "date");
    columnTypeMap.put("time", "time(6)");
    columnTypeMap.put("datetime", "timestamp(6)");
    columnTypeMap.put("datetime2", "timestamp(6)");
    columnTypeMap.put("datetimeoffset", "timestamp_ltz(6)");
    columnTypeMap.put("smalldatetime", "timestamp(6)");
    columnTypeMap.put("year", "int");

    // 字符串类型映射
    columnTypeMap.put("char", "char(2147483647)");
    columnTypeMap.put("nchar", "varchar(2147483647)");
    columnTypeMap.put("varchar", "varchar(2147483647)");
    columnTypeMap.put("nvarchar", "varchar(2147483647)");
    columnTypeMap.put("text", "string");
    columnTypeMap.put("ntext", "string");
    columnTypeMap.put("xml", "string");

    // 二进制类型映射
    columnTypeMap.put("binary", "bytes");
    columnTypeMap.put("varbinary", "bytes");
    columnTypeMap.put("blob", "bytes");
    columnTypeMap.put("image", "bytes");
    columnTypeMap.put("timestamp", "bytes");

    // 其他类型映射
    columnTypeMap.put("boolean", "boolean");
    columnTypeMap.put("enum", "string");
    columnTypeMap.put("set", "string");
    columnTypeMap.put("uniqueidentifier", "string");
    columnTypeMap.put("sql_variant", "string");

    return columnTypeMap;
  }

  @Override
  public String convertColumnType(Map<String, String> flinkColumnTypeMap, String sourceColumnType) {
    return ColumnTypeConverter.super.convertColumnType(flinkColumnTypeMap, sourceColumnType);
  }
}
